package com.ygp.dtp.core;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.ygp.dtp.common.config.properties.DtpProperties;
import com.ygp.dtp.common.config.threadpool.ThreadPoolProperties;
import com.ygp.dtp.common.constant.DynamicTpConstant;
import com.ygp.dtp.common.dto.DynamicThreadPoolMainProperty;
import com.ygp.dtp.common.dto.NotifyItem;
import com.ygp.dtp.common.enums.NotifyTypeEnum;
import com.ygp.dtp.common.enums.QueueTypeEnum;
import com.ygp.dtp.common.enums.RejectedTypeEnum;
import com.ygp.dtp.common.exception.DtpException;
import com.ygp.dtp.common.queue.VariableLinkedBlockingQueue;
import com.ygp.dtp.core.context.DynamicThreadPoolContext;
import com.ygp.dtp.core.context.DynamicThreadPoolContextHolder;
import com.ygp.dtp.core.convert.ExecutorConverter;
import com.ygp.dtp.core.notify.AlarmLimiter;
import com.ygp.dtp.core.notify.NotifyHelper;
import com.ygp.dtp.core.reject.factory.RejectedCountableFactory;
import com.ygp.dtp.core.support.DtpCreator;
import com.ygp.dtp.core.thread.DynamicThreadPoolExecutor;
import com.ygp.dtp.core.thread.ThreadPoolBuilder;
import com.ygp.dtp.equator.Equator;
import com.ygp.dtp.equator.handler.GetterBaseEquator;
import com.ygp.dtp.equator.pojo.FieldInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/**
 * @description: 核心注册表，保存所有已注册的动态线程池执行器。
 * @motto: 代码源于生活，高于生活艺术
 * @author: zhouhengzhe
 * @date: 2022/8/4 17:19
 * @since 1.0
 **/
@Slf4j
@Component
public class DynamicThreadPoolRegistry implements InitializingBean {

    @Resource
    private static RejectedCountableFactory rejectedCountableFactory;

    private static final ExecutorService NOTIFY_EXECUTOR = DtpCreator.createWithTtl(DynamicTpConstant.DTP_NOTIFY);


    private static final Map<String, DynamicThreadPoolExecutor> DTP_REGISTRY = Maps.newConcurrentMap();

    /**
     * getter属性对比器
     */
    private static final Equator EQUATOR = new GetterBaseEquator();

    @Resource
    private static DtpProperties dtpProperties;

    /**
     * 注册动态线程池（用于springboot自动装箱）
     *
     * @param executor 新的线程池
     * @param source   调用注册方法的来源
     */
    public static void register(DynamicThreadPoolExecutor executor, String source) {
        log.info("动态数据源注册成功，source（来源）: {}, executor（线程池）: {}", source, ExecutorConverter.convert(executor));
        DTP_REGISTRY.put(executor.getThreadPoolName(), executor);
    }

    public static DynamicThreadPoolExecutor getExecutor(String name) {
        DynamicThreadPoolExecutor dynamicThreadPoolExecutor = DTP_REGISTRY.get(name);
        if (Objects.isNull(dynamicThreadPoolExecutor)) {
            log.error("找不到指定的动态线程池, name（线程名）: {}", name);
            throw new DtpException("找不到指定的动态线程池, name（线程名）: " + name);
        }
        return dynamicThreadPoolExecutor;
    }

    /**
     * 获取所有配置的动态线程池名称
     *
     * @return
     */
    public static List<String> listAllDtpNames() {
        return Lists.newArrayList(DTP_REGISTRY.keySet());
    }

    public static void refresh(DtpProperties bindDtpProperties) {
        if (Objects.isNull(bindDtpProperties) || CollectionUtils.isEmpty(bindDtpProperties.getExecutors())) {
            log.warn("动态线程池重新刷新，线程池执行器配置为空！");
            return;
        }
        bindDtpProperties.getExecutors().forEach(executor -> {
            DynamicThreadPoolExecutor threadPoolExecutor = DTP_REGISTRY.get(executor.getThreadPoolName());
            if (Objects.isNull(threadPoolExecutor)) {
                log.warn("动态线程池刷新，找不到指定的执行器，name:{}", executor.getThreadPoolName());
                return;
            }
            refresh(threadPoolExecutor, executor);
        });
    }

    /**
     * 刷新指定线程池的指定配置
     *
     * @param threadPoolExecutor   线程池
     * @param threadPoolProperties 线程池配置
     */
    private static void refresh(DynamicThreadPoolExecutor threadPoolExecutor, ThreadPoolProperties threadPoolProperties) {
        //旧配置
        DynamicThreadPoolMainProperty oldProperties = ExecutorConverter.convert(threadPoolExecutor);
        //准备刷新新配置
        doRefresh(threadPoolExecutor, threadPoolProperties);
        //新配置
        DynamicThreadPoolMainProperty newProperties = ExecutorConverter.convert(threadPoolExecutor);
        if (Objects.equals(oldProperties, newProperties)) {
            log.warn("动态线程池：{},属性没有改变", threadPoolExecutor.getThreadPoolName());
            return;
        }
        List<FieldInfo> diffFields = EQUATOR.getDiffFields(oldProperties, newProperties);
        List<String> fieldNames = diffFields.stream().map(FieldInfo::getFieldName).collect(Collectors.toList());
        DynamicThreadPoolContext contextWrapper = DynamicThreadPoolContext.builder()
                .dynamicThreadPoolExecutor(threadPoolExecutor)
                .platforms(dtpProperties.getPlatforms())
                .notifyItem(NotifyHelper.getNotifyItem(threadPoolExecutor, NotifyTypeEnum.CHANGE))
                .build();
        DynamicThreadPoolContextHolder.set(contextWrapper);
        NOTIFY_EXECUTOR.execute(() -> {
            //发送告警信息
            NotifyHelper.getInstance().sendNotify(oldProperties, newProperties);
        });
        //打印日志
        printLog(threadPoolExecutor, oldProperties, newProperties, fieldNames);

    }

    /**
     * 打印日志信息
     *
     * @param threadPoolExecutor
     * @param oldProperties
     * @param newProperties
     * @param fieldNames
     */
    private static void printLog(DynamicThreadPoolExecutor threadPoolExecutor,
                                 DynamicThreadPoolMainProperty oldProperties,
                                 DynamicThreadPoolMainProperty newProperties,
                                 List<String> fieldNames) {
        log.info("DynamicTp [{}] refreshed end, changed keys: {}, corePoolSize: [{}], maxPoolSize: [{}], " +
                        "queueType: [{}], queueCapacity: [{}], keepAliveTime: [{}], rejectedType: [{}], " +
                        "allowsCoreThreadTimeOut: [{}]",
                threadPoolExecutor.getThreadPoolName(),
                fieldNames,
                String.format(DynamicTpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.getCorePoolSize(), newProperties.getCorePoolSize()),
                String.format(DynamicTpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.getMaxPoolSize(), newProperties.getMaxPoolSize()),
                String.format(DynamicTpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.getQueueType(), newProperties.getQueueType()),
                String.format(DynamicTpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.getQueueCapacity(), newProperties.getQueueCapacity()),
                String.format("%ss => %ss", oldProperties.getKeepAliveTime(), newProperties.getKeepAliveTime()),
                String.format(DynamicTpConstant.PROPERTIES_CHANGE_SHOW_STYLE, RejectedTypeEnum.formatRejectName(newProperties.getRejectType()),
                        RejectedTypeEnum.formatRejectName(oldProperties.getRejectType())),
                String.format(DynamicTpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.isAllowCoreThreadTimeOut(),
                        newProperties.isAllowCoreThreadTimeOut()));
    }

    /**
     * 准备把新配置刷新进去
     *
     * @param threadPoolExecutor
     * @param threadPoolProperties
     */
    private static void doRefresh(DynamicThreadPoolExecutor threadPoolExecutor, ThreadPoolProperties threadPoolProperties) {
        //核心线程数不能大于最大线程数（1.8 bug）
        if (threadPoolProperties.getCorePoolSize() > threadPoolProperties.getMaximumPoolSize()) {
            throw new IllegalArgumentException();
        }
        //更新核心线程数
        if (threadPoolProperties.getCorePoolSize() >= 0) {
            threadPoolExecutor.setCorePoolSize(threadPoolProperties.getCorePoolSize());
        }
        //更新最大线程数
        if (threadPoolProperties.getMaximumPoolSize() >= 0) {
            threadPoolExecutor.setMaximumPoolSize(threadPoolProperties.getMaximumPoolSize());
        }
        //更新活跃时间与单位
        if (threadPoolProperties.getKeepAliveTime() > 0 && Objects.nonNull(threadPoolProperties.getUnit())) {
            threadPoolExecutor.setKeepAliveTime(threadPoolProperties.getKeepAliveTime(), threadPoolProperties.getUnit());
        }
        //更新是否允许核心线程超时
        threadPoolExecutor.allowCoreThreadTimeOut(threadPoolProperties.isAllowCoreThreadTimeOut());

        //更新拒绝策略
        String originRejectHandlerName = threadPoolExecutor.getRejectHandlerName();
        String newRejectedHandlerType = threadPoolProperties.getRejectedHandlerType();
        if (StringUtils.isNotBlank(newRejectedHandlerType) && !originRejectHandlerName.contains(newRejectedHandlerType)) {
            threadPoolExecutor.setRejectedExecutionHandler(rejectedCountableFactory.getRejected(newRejectedHandlerType));
        }

        //更新阻塞队列
        if (threadPoolProperties.getQueueCapacity() > 0
                && Objects.equals(threadPoolProperties.getQueueType(), QueueTypeEnum.VARIABLE_LINKED_BLOCKING_QUEUE.getName())) {
            BlockingQueue<Runnable> queue = threadPoolExecutor.getQueue();
            if (queue instanceof VariableLinkedBlockingQueue) {
                ((VariableLinkedBlockingQueue<Runnable>) queue).setCapacity(threadPoolProperties.getQueueCapacity());
            } else {
                log.error("动态线程池刷新，阻塞队列容量无法重置，线程池名：{}，队列类型：{}",
                        threadPoolExecutor.getThreadPoolName(), threadPoolExecutor.getQueueName());
            }
        }
        if (CollectionUtils.isEmpty(threadPoolProperties.getNotifyItems())) {
            threadPoolExecutor.setNotifyItems(NotifyItem.getDefaultNotifyItems());
            return;
        }
        NotifyHelper.setExecutorNotifyItems(threadPoolExecutor, dtpProperties, threadPoolProperties);
        threadPoolExecutor.setNotifyItems(threadPoolProperties.getNotifyItems());
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        if (CollectionUtils.isEmpty(dtpProperties.getExecutors())) {
            return;
        }
        dtpProperties.getExecutors().forEach(i -> {
            ThreadPoolBuilder.builder()
                    .corePoolSize(i.getCorePoolSize())
                    .maximumPoolSize(i.getMaximumPoolSize())
                    .keepAliveTime(i.getKeepAliveTime())
                    .workQueue(i.getQueueType(), i.getQueueCapacity(), i.isFair())
                    .rejectedExecutionHandler(i.getRejectedHandlerType())
                    .threadFactory(i.getThreadNamePrefix())
                    .allowCoreThreadTimeOut(i.isAllowCoreThreadTimeOut())
                    .threadPoolName(i.getThreadPoolName())
                    .notifyItems(i.getNotifyItems())
                    .buildDynamic();
        });
        DTP_REGISTRY.forEach((k, v) -> {
            NotifyHelper.fillNotifyItems(dtpProperties.getPlatforms(), v.getNotifyItems());
            v.getNotifyItems().forEach(i -> {
                AlarmLimiter.initAlarmLimiter(k, v);
            });
        });
    }
}